package groupOne.app.DWD.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import groupOne.app.BaseAppOneStream;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import groupOne.util.FlinkSinkUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import static groupOne.common.Constant.TOPIC_ODS_LOG;

public class Dwd_01_DwdBaseLogApp extends BaseAppOneStream {
    public static void main(String[] args) {
        new Dwd_01_DwdBaseLogApp().init(2001,2,"Dwd_01_DwdBaseLogApp",TOPIC_ODS_LOG);
    }
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        OutputTag<String> actionTag = new OutputTag<String>("action"){};
        OutputTag<String> errTag = new OutputTag<String>("err"){};
        OutputTag<String> displayTag = new OutputTag<String>("display"){};
        OutputTag<String> pagTag = new OutputTag<String>("actpagion"){};
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideo"){};

        SingleOutputStreamOperator<String> startStream = stream
                .flatMap(new FlatMapFunction<String, JSONObject>() {
                    @Override
                    public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(s);
                            collector.collect(jsonObject);
                        } catch (Exception e) {
                            System.out.println("数据格式不对");
                        }
                    }
                })
                .keyBy(a -> a.getJSONObject("common").getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, String>() {

                    private ValueState<String> isNewState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        isNewState = getRuntimeContext().getState(new ValueStateDescriptor<String>("isNewState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) throws Exception {
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject start = jsonObject.getJSONObject("start");
                        JSONObject page = jsonObject.getJSONObject("page");
                        JSONObject appVideo = jsonObject.getJSONObject("appVideo");
                        String isNew = common.getString("is_new");

                        Long ts = jsonObject.getLong("ts");
                        //新老访客标记修复
                        if (isNewState == null) {
                            if ("0".equals(isNew)) {
                                isNewState.update(AtguiguUtil.toDate(ts - 24 * 60 * 60 * 1000));
                            } else {
                                isNewState.update(AtguiguUtil.toDate(ts));
                            }
                        } else {
                            if ("1".equals(isNew)) {
                                if (AtguiguUtil.toDate(ts) != isNewState.value()) {
                                    common.put("is_new", "0");
                                }
                            }
                        }

                        //启动日志输出到主流
                        if (start != null) {
                            collector.collect(jsonObject.toJSONString());
                        } else if(appVideo != null){
                            //appVideo
                            appVideo.putAll(common);
                            appVideo.put("ts",ts);
                            context.output(appVideoTag,appVideo.toJSONString());
                        }else {
                            //其他日志
                            JSONArray actions = jsonObject.getJSONArray("actions");
                            JSONArray displays = jsonObject.getJSONArray("displays");
                            //行动日志
                            if (actions != null) {
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    action.putAll(common);
                                    action.putAll(page);
                                    context.output(actionTag,action.toJSONString());
                                }
                                jsonObject.remove("actions");
                            }

                            //displays
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    display.putAll(common);
                                    display.putAll(page);
                                    display.put("ts",ts);
                                    context.output(displayTag,display.toJSONString());
                                }
                                jsonObject.remove("displays");
                            }
                            //错误
                            if (jsonObject.containsKey("err")) {
                                context.output(errTag,jsonObject.toJSONString());
                                jsonObject.remove("err");
                            }
                            //页面日志
                            if (page != null) {
                                context.output(pagTag,jsonObject.toJSONString());
                            }
                       }
                    }
                });
        //写出到kafka
        startStream.addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        startStream.getSideOutput(actionTag).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        startStream.getSideOutput(displayTag).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        startStream.getSideOutput(pagTag).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        startStream.getSideOutput(errTag).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        startStream.getSideOutput(appVideoTag).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PLAY_VIDEO));
    }
}
